package com.xunmeng.chapter4;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

/**
 * 服务端采用 boss - worker 模式
 */
@Slf4j
public class WorkServer {

    public static void main(String[] args) throws IOException {// boss
        ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
        // 创建selector
        Selector selector=Selector.open();
        serverSocketChannel.configureBlocking(false);// 关闭阻塞
        // 绑定地址
        serverSocketChannel.bind(new InetSocketAddress(8080));
        // 监听事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT,null);

        Worker[] workers=new Worker[Runtime.getRuntime().availableProcessors()];
        for(int i=0;i<workers.length;i++) workers[i] = new Worker("worker-"+i);
        AtomicInteger index=new AtomicInteger();
        while (true){
            selector.select();
            // 获取客户端连接
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey ssckey = iterator.next();
                iterator.remove();
                if(ssckey.isAcceptable()){
                    ServerSocketChannel ssc = (ServerSocketChannel) ssckey.channel();
                    SocketChannel socketChannel = ssc.accept();
                    socketChannel.configureBlocking(false);
                    // 将该socketChannel绑定到worker上
                    log.info("注册客户端【{}】前...",socketChannel.getRemoteAddress());
                    workers[index.getAndIncrement()%workers.length].register(socketChannel,SelectionKey.OP_READ,null);
                    log.info("注册客户端【{}】后...",socketChannel.getRemoteAddress());
                }
            }
        }
    }

    private static class Worker implements Runnable{// 工作者

        private String name;// 名称
        private Selector selector;// selector
        private Thread thread;// 线程
        private volatile boolean flag=false;
        private ConcurrentLinkedQueue<Function<SocketChannel,Object>> queue=new ConcurrentLinkedQueue<>();


        // 构造函数
        public Worker(String name){
            this.name=name;
        }

        public void register(SocketChannel sc,int opts,Object att) throws IOException {
            if(!flag){
                this.thread=new Thread(this);
                selector=Selector.open();
                thread.start();
                flag=true;
            }
            // 将 sc 放到 队列中，两个线程之间的通信用队列

//            queue.add((a)->{
//                try {
//                    sc.register(selector,opts,att);
//                } catch (ClosedChannelException e) {
//                    e.printStackTrace();
//                }
//                return null;
//            });
            selector.wakeup();
            sc.register(selector,opts,att);
        }

        @Override
        public void run() {
            while (true){
                try {
                    selector.select();
//                    while (!queue.isEmpty()){
//                        Function<SocketChannel, Object> poll = queue.poll();
//                        poll.apply(null);
//                    }
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        // 判断当前key的类型
                        if (key.isReadable()){
                            SocketChannel sc = (SocketChannel) key.channel();
                            // 读取内容
                            ByteBuffer buffer=ByteBuffer.allocate(16);
                            try {
                                int read = sc.read(buffer);
                                if (read==-1){// 正常关闭
                                    key.cancel();
                                }
                                buffer.flip();
                                log.info("服务端【{}】接收到客户端【{}】的消息为：{}",name,sc.getRemoteAddress(), StandardCharsets.UTF_8.decode(buffer));
                            }catch (Exception e){
                                key.cancel();
                            }

                        }
                    }


                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        }
    }
}
